package com.dap.utils;

import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.LinkedBlockingQueue;

public class AsyncQueue {
    private static final Logger logger = LoggerFactory.getLogger(AsyncQueue.class);

    public interface AsyncQueueConsumer {
        void consumeAsyncQueueContent(Object content);
    }
    private static class QueueElement {
        public final AsyncQueueConsumer asyncQueueConsumer;
        public final Object content;
        public final Long earliestConsumeTime;  // 最早消费时间, 在这个时间之前不会被消费; 改属性为null则没有限制
        public QueueElement(AsyncQueueConsumer asyncQueueConsumer, Object content, Long earliestConsumeTime) {
            this.asyncQueueConsumer = asyncQueueConsumer;
            this.content = content;
            this.earliestConsumeTime = earliestConsumeTime;
        }
    }

    private final LinkedBlockingQueue<QueueElement> queue = new LinkedBlockingQueue<>(20480);
    public boolean enqueueMessage(AsyncQueueConsumer asyncQueueConsumer, Object content, Long earliestConsumeTime) {
        if (null == asyncQueueConsumer) {
            logger.info("enqueue failed, asyncQueueConsumer is null");
            throw new RuntimeException("asyncQueueConsumer is null");
        }
        QueueElement queueElement = new QueueElement(asyncQueueConsumer, content, earliestConsumeTime);

        // offer方法在添加元素时，如果发现队列已满无法添加的话，会直接返回false
        if (!queue.offer(queueElement)) {
            logger.info("enqueue failed, will send later, message={}", JSON.toJSONString(queueElement));
            return false;
        } else {
            return true;
        }
    }


    private static void waitUntilTime(final long time) {
        while (true) {
            long diff = time - System.currentTimeMillis();
            if (diff > 0) {
                try {
                    Thread.sleep(diff);
                } catch (Exception e) {
                    logger.info("", e);
                }
            } else {
                break;
            }
        }
    }

    private Thread thread;
    private volatile boolean isThreadRun = true;
    public AsyncQueue() {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                while (isThreadRun) {
                    try {
                        // take:若队列为空，发生阻塞，等待有元素
                        QueueElement queueElement = queue.take();
                        if (queueElement != null) {
                            if (queueElement.earliestConsumeTime != null) {
                                waitUntilTime(queueElement.earliestConsumeTime);
                            }
                            queueElement.asyncQueueConsumer.consumeAsyncQueueContent(queueElement.content);
                        }
                    } catch (Exception e) {
                        logger.warn("", e);
                    }
                }
            }
        };

        thread = new Thread(runnable, "AsyncQueueThread"+this);
        thread.start();
    }

    public void close() {
        isThreadRun = false;
    }
}
